package com.atguigu.flink.chapter10.query;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
 窗口表值函数（TVF）聚合
 */
public class QueryWindowTVFDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        // 在 ddl 中定义时间属性
        tEnv.executeSql("create table sensor(" +
                " id string, " +
                " ts bigint, " +
                " vc int, " +
                " et as to_timestamp_ltz(ts, 3)," +
                " watermark for et as et - interval '1' second " +
                ")with(" +
                "  'connector' = 'filesystem', " +
                "  'path' = 'input/sensor.txt', " +
                "  'format' = 'csv' " +
                ")");


        tEnv.sqlQuery("select " +
                "id,window_start,window_end, " +
                "sum(vc) vc_sum" +
                "from table(tumble(table sensor,descriptor(et),interval '5' second))" +
                " group by id,window_start,window_end")
                .execute()
                .print();



        tEnv.sqlQuery("select " +
                        "id,window_start,window_end, " +
                        "sum(vc) vc_sum" +
                        "from table(tumble(table sensor,descriptor(et),interval '5' second))" +
                        " group by id,window_start,window_end")
                .execute()
                .print();


        tEnv.sqlQuery("select " +
                        "id,window_start,window_end, " +
                        "sum(vc) vc_sum" +
                        "from table(hop(table sensor,descriptor(et),interval '2' second,interval '6' second))" +
                        " group by id,window_start,window_end")
                .execute()
                .print();


        //累积窗口
        //每隔1h 计算下当天的pv

        tEnv.sqlQuery("select " +
                        "id,window_start,window_end, " +
                        "sum(vc) vc_sum" +
                        "from table(cumulate(table sensor,descriptor(et),interval '5' second,interval '20' second))" +
                        " group by id,window_start,window_end")
                .execute()
                .print();
    }
}



//public class QueryWindowTVFDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        //        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//
//        // 在 ddl 中定义时间属性
//        tEnv.executeSql("create table sensor(" +
//                " id string, " +
//                " ts bigint, " +
//                " vc int, " +
//                " et as to_timestamp_ltz(ts, 3)," +
//                " watermark for et as et - interval '1' second " +
//                ")with(" +
//                "  'connector' = 'filesystem', " +
//                "  'path' = 'input/sensor.txt', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        tEnv.sqlQuery("select " +
//                "id,window_start,window_end, " +
//                "sum(vc) vc_sum" +
//                "from table(tumble(table sensor,descriptor(et),interval '5' second))")
//                .execute()
//                .print();
//
//        tEnv.sqlQuery("select " +
//                        "id,window_start,window_end, " +
//                        "sum(vc) vc_sum" +
//                        "from table(hop(table sensor,descriptor(et),interval '2' second,interval '6' second))")
//                .execute()
//                .print();
//
//
//        tEnv.sqlQuery("select " +
//                        "id,window_start,window_end, " +
//                        "sum(vc) vc_sum" +
//                        "from table(cumulate(table sensor,descriptor(et),interval '5' second,interval '20' second))")
//                .execute()
//                .print();
//    }
//}